#include <iostream>
#include <vector>
#include <string>
#include <algorithm>
#include <unordered_map>
#include <functional>
#include <regex>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <memory>
#include <any>
#include <assert.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#include <time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>

#define __BUFFER_DEFAULT_SIZE_ 1024

// 日志宏
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF

#define LOG(level, format, ...)                                                             \
    do                                                                                      \
    {                                                                                       \
        if (level < LOG_LEVEL)                                                              \
            break;                                                                          \
        time_t t = time(nullptr);                                                           \
        struct tm *_timer = localtime(&t);                                                  \
        char tmp[64] = {0};                                                                 \
        strftime(tmp, 63, "%H:%M:%S", _timer);                                              \
        fprintf(stdout, "[%s %s:%d] " format "\n", tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define LOG_INFO(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define LOG_DEBUG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define LOG_ERROR(format, ...) LOG(ERR, format, ##__VA_ARGS__)

// 缓冲区类
class Buffer
{
public:
    Buffer() : _buffer(__BUFFER_DEFAULT_SIZE_), _RIdx(0), _WIdx(0)
    {
    }

    ~Buffer()
    {
    }

    char *WritePosition() // 获取当前写入位置地址
    {
        return &(*(_buffer.begin() + _WIdx));
    }
    char *ReadPosition() // 获取当前读取位置地址
    {
        return &(*(_buffer.begin() + _RIdx));
    }

    const char *ReadPosition() const // 获取当前读取位置地址
    {
        return &(*(_buffer.begin() + _RIdx));
    }

    uint64_t TailFreeSize() // 获取末尾的空闲区域大小
    {
        return _buffer.size() - _WIdx;
    }
    uint64_t HeadFreeSize() // 获取头部空闲区域大小
    {
        return _RIdx;
    }
    uint64_t ReadbleSize() const // 获取当前可读数据大小
    {
        return _WIdx - _RIdx;
    }
    void MoveReadOffest(uint64_t _len) // 将读位置向后偏移
    {
        if (_len > ReadbleSize())
        {
            return;
        }
        _RIdx += _len;
    }
    void MoveWriteOffest(uint64_t _len) // 将写位置向后偏移
    {
        if (_len > TailFreeSize())
        {
            return;
        }
        _WIdx += _len;
    }
    void EnsureWriteSpace(uint64_t _len) // 确保可写空间足够 //如果当前空闲位置足够,则向前移动数据 否则扩容
    {
        if (_len <= TailFreeSize())
        {
            return;
        }
        if (_len <= HeadFreeSize() + TailFreeSize()) // 1.当前空闲区域足够
        {
            int _res = ReadbleSize();
            std::copy(ReadPosition(), ReadPosition() + _res, &(*_buffer.begin()));
            _RIdx = 0;
            _WIdx = _RIdx + _res;
        }
        else
            _buffer.resize(_WIdx + _len); // 2.不够 则扩容
    }
    void Write(const void *data, uint64_t _len) // 写入数据
    {
        if (_len == 0)
            return;
        EnsureWriteSpace(_len);
        const char *d = (const char *)data;
        std::copy(d, d + _len, WritePosition());
    }

    void WriteAndPush(const void *data, uint64_t _len) // 写入数据并移动写偏移
    {
        Write(data, _len);
        MoveWriteOffest(_len);
    }

    void WriteString(const std::string &_str) // 写入字符串
    {
        Write(_str.c_str(), _str.size());
    }

    void WriteStringAndPush(const std::string _str) // 写入字符串并移动写偏移
    {
        WriteString(_str);
        MoveWriteOffest(_str.size());
    }

    void WriteBuffer(const Buffer &_buf) // 写入Buffer
    {
        Write(_buf.ReadPosition(), _buf.ReadbleSize());
    }

    void WriteBufferAndPush(const Buffer &_buf) // 写入Buffer并移动写偏移
    {
        WriteBuffer(_buf);
        MoveWriteOffest(_buf.ReadbleSize());
    }

    void Read(void *buffer, uint64_t _len) // 读取数据
    {
        assert(_len <= ReadbleSize());
        std::copy(ReadPosition(), ReadPosition() + _len, (char *)buffer);
    }

    void ReadAndPop(void *buffer, uint64_t _len) // 读取数据并移动读偏移
    {
        Read(buffer, _len);
        MoveReadOffest(_len);
    }

    std::string ReadAsString(uint64_t _len) // 读取数据并作为字符串返回
    {
        std::string _str;
        _str.resize(_len);
        Read(&_str[0], _len);
        return _str;
    }

    std::string ReadAsStringAndPop(uint64_t _len) // 读取数据作为字符串返回并移动读偏移
    {
        std::string _str = ReadAsString(_len);
        MoveReadOffest(_len);
        return _str;
    }

    char *FindCRLF() // 返回'\n'的位置 http
    {
        void *ret = memchr(ReadPosition(), '\n', ReadbleSize());
        return (char *)ret;
    }

    std::string GetLine() // 返回一行数据 以'\n'为标志 包括\n http
    {
        char *pos = FindCRLF();
        if (pos == nullptr)
            return "";
        std::string _str = ReadAsString(pos - ReadPosition() + 1);
        return _str;
    }

    std::string GetLineAndPop() // 返回一行以'\n'为标志的字符串并移动读偏移
    {
        std::string _str = GetLine();
        MoveReadOffest(_str.size());
        return _str;
    }

    void Clear() // 清空缓冲区
    {
        _WIdx = _RIdx = 0;
    }

private:
    std::vector<char> _buffer; // 缓冲区
    uint64_t _RIdx;            // 相对读偏移量
    uint64_t _WIdx;            // 相对写偏移量
};

#define _BACK_LOG_ 1024
#define _DEFAULT_SOCKFD_ -1

// 套接字类
class Socket
{
public:
    // 构造
    Socket() : _sockfd(_DEFAULT_SOCKFD_)
    {
    }
    // 构造
    Socket(int sockfd) : _sockfd(sockfd)
    {
    }
    // 析构
    ~Socket()
    {
        Close();
    }

    // 创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            LOG_ERROR("create socket error");
            return false;
        }
        int _opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &_opt, sizeof(_opt)); // 设置ip地址重用
        return true;
    }
    // 绑定地址信息
    bool Bind(const std::string &_ip, uint16_t _port)
    {
        struct sockaddr_in _addr;
        _addr.sin_addr.s_addr = inet_addr(_ip.c_str());
        _addr.sin_family = AF_INET;
        _addr.sin_port = htons(_port);
        socklen_t _len = sizeof(_addr);
        int _ret = bind(_sockfd, (const struct sockaddr *)&_addr, _len);
        if (_ret < 0)
        {
            LOG_ERROR("bind socket error");
            return false;
        }
        return true;
    }
    // 监听
    bool Listen(int _backlog = _BACK_LOG_)
    {
        int _ret = listen(_sockfd, _backlog);
        if (_ret < 0)
        {
            LOG_ERROR("Listen socket error");
            return false;
        }
        return true;
    }
    // 客户端向服务器发起连接
    bool Connect(const std::string &_ip, uint16_t _port)
    {
        struct sockaddr_in _addr;
        _addr.sin_addr.s_addr = inet_addr(_ip.c_str());
        _addr.sin_family = AF_INET;
        _addr.sin_port = htons(_port);

        int n = connect(_sockfd, (const struct sockaddr *)&_addr, sizeof(_addr));
        if (n < 0)
        {
            LOG_ERROR("connect server error");
            return false;
        }
        return true;
    }
    // 获取新连接
    int Accept()
    {
        struct sockaddr_in _addr;
        memset(&_addr, 0, sizeof(_addr));
        socklen_t _len = sizeof(_addr);
        int _Sockfd = accept(_sockfd, (struct sockaddr *)&_addr, &_len);
        if (_Sockfd < 0)
        {
            LOG_ERROR("accept client error");
            return -1;
        }

        return _Sockfd;
    }
    // 接收数据
    ssize_t Recv(void *_buffer, size_t _len, int flag = 0)
    {
        ssize_t n = recv(_sockfd, _buffer, _len, flag);
        if (n <= 0)
        {
            // EAGAIN:表示当前缓冲区里已经没有数据可读了
            // EINTR:表示recv突然被连接打断了
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0; // 表示当前没有在缓冲区读到数据
            }
            LOG_ERROR("socket recv error");
            return -1; // 表示读取出错
        }
        return n;
    }

    ssize_t NonBlockRecv(void *_buffer, size_t _len)
    {
        return Recv(_buffer, _len, MSG_DONTWAIT);
    }

    // 发送数据
    ssize_t Send(void *_buffer, size_t _len, int flag = 0)
    {
        ssize_t _ret = send(_sockfd, _buffer, _len, flag);
        if (_ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0; // 表示当前没有在缓冲区读到数据
            }

            LOG_ERROR("socket send error");
            return -1; // 发送失败
        }
        return _ret; // 实际发送的数据长度
    }

    ssize_t NonBlockSend(void *_buffer, size_t _len)
    {
        return Send(_buffer, _len, MSG_DONTWAIT);
    }

    // 关闭套接字
    void Close()
    {
        close(_sockfd);
        _sockfd = -1;
    }
    // 创建一个服务端连接
    bool CreateServer(uint16_t _port, const std::string &_ip = "0.0.0.0", int flag = 0)
    {
        if (Create() == false)
            return false; // 1.创建套接字
        if (flag)
            SetNonBlock(); // 2.设置非阻塞
        if (Bind(_ip, _port) == false)
            return false; // 绑定地址信息
        SetSocketReuseAddress();
        if (Listen() == false)
            return false; // 监听

        return true;
    }
    // 创建一个客户端连接
    bool CreateClient(uint16_t _port, const std::string &_ip)
    {
        // 1.创建套接字 2.连接服务器
        if (Create() == false)
            return false;
        if (Connect(_ip, _port) == false)
            return false;

        return true;
    }
    // 开启ip和port重用
    void SetSocketReuseAddress()
    {
        int _opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &_opt, sizeof(_opt)); // 设置ip地址重用
        int _val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &_val, sizeof(_val)); // 设置port端口重用
    }
    // 设置套接字为非阻塞
    void SetNonBlock()
    {
        std::cout << "SetNonBlock()" << std::endl;
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }

    int GetFd()
    {
        return _sockfd;
    }

private:
    int _sockfd;
};

// Channel 类

using Action = std::function<void()>;

class Poller;
class EventLoop;
class Channel
{

public:
    Channel(int sockfd, EventLoop *loop) : _sockfd(sockfd), _events(0), _revents(0), _loop(loop)
    {
	    if (_loop == nullptr) {
		    LOG_INFO("channel初始化的时候传入的loop是空");
	    }
    }

    ~Channel()
    {
    }

    void SetReadAction(const Action &_func) // 设置可读事件回调
    {
        _read_action = _func;
    }
    void SetWriteAction(const Action &_func) // 设置可写事件回调
    {
        _write_action = _func;
    }
    void SetCloseAction(const Action &_func) // 设置关闭事件回调
    {
        _close_action = _func;
    }
    void SetAnyAction(const Action &_func) // 设置任意事件触发回调
    {
        _any_action = _func;
    }
    void SetErrorAction(const Action &_func) // 设置错误事件触发回调
    {
        _error_action = _func;
    }

    void SetRevents(uint32_t events)
    {
        _revents = events;
    }

    bool ReadAble() // 当前是否监控了可读
    {
        return _events & EPOLLIN;
    }
    bool WriteAble() // 当前是否监控了可写
    {
        return _events & EPOLLOUT;
    }
    void EnableRead() // 启动读事件监控
    {
        _events |= EPOLLIN;
        Update();
    }
    void EnableWrite() // 启动写事件监控
    {
        _events |= EPOLLOUT;
        Update();
    }
    void DisableRead() // 关闭读事件监控
    {
        _events &= ~EPOLLIN;
        Update();
    }
    void DisableWrite() // 关闭写事件监控
    {
        _events &= ~EPOLLOUT;
        Update();
    }
    void DisableAll() // 关闭所有事件监控
    {
        _events = 0;
        Remove();
    }
    void Remove(); // 移除监控

    void Update(); // 更新监控事件

    void EventAction() // 事件处理
    {
        if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if (_read_action)
                _read_action();
            // if (_any_action)
            //     _any_action();
        }

        if (_revents & EPOLLOUT)
        {
            if (_write_action)
                _write_action();
            if (_any_action)
                _any_action();
        }

        else if (_revents & EPOLLERR)
        {
            if (_any_action)
                _any_action();

            if (_error_action)
                _error_action();
        }

        else if (_revents & EPOLLHUP)
        {
            if (_any_action)
                _any_action();

            if (_close_action)
                _close_action();
        }
    }

    int Fd()
    {
        return _sockfd;
    }

    uint32_t GetEvents()
    {
        return _events;
    }

private:
    uint32_t _events;  // 当前需要监控的事件
    uint32_t _revents; // 当前已经触发的事件
    int _sockfd;
    EventLoop *_loop; // 回指

    Action _read_action;  // 触发可读事件需要执行的回调函数
    Action _write_action; // 触发可写事件需要执行的回调函数
    Action _close_action; // 连接关闭需要执行的回调函数
    Action _any_action;   // 任意事件触发都需要执行的回调函数
    Action _error_action; // 错误事件触发需要执行的回调函数
};

#define _MAX_EVENTS 1024

class Poller
{
private:
    bool IsExit(Channel *_channel)
    {
        auto _iter = _channels.find(_channel->Fd());
        if (_iter == _channels.end())
            return false;

        return true;
    }

    void Update(Channel *_channel, int op)
    {
        // std::cout << "Update : " << _channel->Fd() << std::endl;
        struct epoll_event _ev;
        _ev.data.fd = _channel->Fd();
        _ev.events = _channel->GetEvents();
        int _ret = epoll_ctl(_epfd, op, _channel->Fd(), &_ev);
        if (_ret < 0)
        {
            LOG_ERROR("epoll ctrl error");
            exit(1);
        }
    }

public:
    Poller() : _epfd(epoll_create(_MAX_EVENTS))
    {
    }

    void UpdateEvents(Channel *_channel) // 存在就更新 不存在就push
    {
        if (IsExit(_channel))
        {
            Update(_channel, EPOLL_CTL_MOD);
        }

        else
        {
            _channels[_channel->Fd()] = _channel;
            Update(_channel, EPOLL_CTL_ADD);
        }
    }

    void RemoveEvents(Channel *_channel) // 清除事件监控
    {
        auto _iter = _channels.find(_channel->Fd());
        if (_iter != _channels.end())
        {
            _channels.erase(_iter);
            Update(_channel, EPOLL_CTL_DEL);
        }
    }

    void Poll(std::vector<Channel *> *_active)
    {
        int _fds = epoll_wait(_epfd, revs, _MAX_EVENTS, -1);
        for (int i = 0; i < _fds; i++)
        {
            auto _iter = _channels.find(revs[i].data.fd);
            if (_iter == _channels.end())
            {
                LOG_ERROR("epoll wait error");
                return;
            }
            _iter->second->SetRevents(revs[i].events); // 返回实际就绪的事件
            _active->push_back(_iter->second);         // 插入活跃的连接
        }
    }

    ~Poller()
    {
    }

private:
    int _epfd;                                    // epoll模型的文件描述符
    struct epoll_event revs[_MAX_EVENTS];         // 用于存放就绪事件的数组
    std::unordered_map<int, Channel *> _channels; // 方便查找和修改Channel
};

using Task = std::function<void()>;
using Release = std::function<void()>;

class TimerTask
{
public:
    TimerTask(uint64_t id, uint32_t timeout, const Task &task)
        : _id(id), _timeout(timeout), _task(task), _cancel(false) // cancel默认为没有取消任务
    {
    }

    void SetRelease(const Release &cb)
    {
        _release = cb;
    }

    uint32_t GetTimeOut()
    {
        return _timeout;
    }

    void Cancel()
    {
        _cancel = true;
    }

    ~TimerTask()
    {
        if (_cancel == false) // 如果没有被取消,就执行任务
            _task();
        _release();
		LOG_INFO("定时任务释放");
    }

private:
    uint64_t _id;      // 定时任务id
    uint32_t _timeout; // 定时任务的超时时间
    Task _task;        // 定时任务执行操作
    Release _release;  // 定时任务释放操作
    bool _cancel;      // 任务是否被取消
};

using _PtrTask = std::shared_ptr<TimerTask>;
using _WeakPtr = std::weak_ptr<TimerTask>;

class EventLoop;
class TimeWheel
{
private:
    static int CreateTimerfd()
    {
        int _timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (_timerfd < 0)
        {
            LOG_ERROR("timerfd create error");
            exit(1);
        }

        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(_timerfd, 0, &itime, NULL);

        return _timerfd;
    }

    int ReadTimerfd()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, 8);
        if (ret < 0)
        {
            perror("read error");
            exit(1);
        }
        return times;
    }

    void Run()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear();
    }

    void RunOnTime()
    {
        int times = ReadTimerfd();
        for(int i = 0;i<times;i++)
        {
            Run();
        }
    }

    void TimerTaskAdd(uint64_t _id, uint32_t _timeout, const Task &_task) // 添加定时任务
    {
        _PtrTask _sptr(new TimerTask(_id, _timeout, _task));
        _sptr->SetRelease(std::bind(&TimeWheel::Remove, this, _id)); // 设置清除方法
        _wheel[(_tick + _timeout) % _capacity].push_back(_sptr);     // 插入到时间轮中

        _timers[_id] = _WeakPtr(_sptr); // 插入到哈希中
    }

    void TimerTaskRefresh(uint64_t _id) // 延迟/刷新定时任务
    {
        auto it = _timers.find(_id);
        if (it == _timers.end())
        {
            return;
        }

        _PtrTask _sptr = it->second.lock();
        _wheel[(_tick + _sptr->GetTimeOut()) % _capacity].push_back(_sptr);
        // 把shared_ptr拷贝之后再次加入_wheel中
        // 使得该位置的TimerTask的shard_ptr的引用计数++,从而延缓了调用析构的时间,达到任务执行时间刷新的效果
    }

    void TimerCancel(int _id)
    {
        auto iter = _timers.find(_id);
        if (iter == _timers.end())
        {
            return;
        }

        _PtrTask _sptr = iter->second.lock();
        if (_sptr) _sptr->Cancel();//做个判断靠谱一些
    }

public:
    TimeWheel(EventLoop *loop, int tick = 0, int capacity = 60)
        : _tick(tick), _capacity(capacity), _wheel(_capacity),
          _timerfd(CreateTimerfd()), _loop(loop), _timerfd_channel(std::make_unique<Channel>(_timerfd, _loop))
    {
        _timerfd_channel->SetReadAction(std::bind(&TimeWheel::RunOnTime, this));
        _timerfd_channel->EnableRead();
    }

    void Remove(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end())
        {
            _timers.erase(it);
        }
    }

    void TimerTaskAddInLoop(uint64_t _id, uint32_t _timeout, const Task &_task);

    void TimerTaskRefreshInLoop(uint64_t _id);

    void TimerCancelInLoop(int _id);

    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end())
        {
            return true;
        }

        return false;
    }

    ~TimeWheel()
    {
    }

private:
    int _tick;                                      // 秒针
    int _capacity;                                  // 表盘最大延迟时间
    std::vector<std::vector<_PtrTask>> _wheel;      // 时钟轮
    std::unordered_map<uint64_t, _WeakPtr> _timers; // 方便寻找TimeTask对象
    int _timerfd;                                   // 引入定时器
    EventLoop *_loop;                               // 回指eventloop
    std::unique_ptr<Channel> _timerfd_channel;      // 管理定时器监控事件
};

using _Func = std::function<void()>;

class EventLoop
{
private:
    void RunAllTask() // 执行所有任务
    {
        std::vector<_Func> _func;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(_func);
        }

        for (auto &e : _func)
        {
            e();
        }
        return;
    }

    static int CreateEventfd() // 创建eventfd
    {
        int evefd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (evefd < 0)
        {
            LOG_ERROR("eventfd error");
            exit(1);
        }
        return evefd;
    }
    void ReadEventfd() // 读取eventfd里面的内容
    {
        uint64_t _res = 0;
        int ret = read(_eventfd, &_res, sizeof(uint64_t));
        if (ret < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
                return;
            LOG_ERROR("eventfd read error");
            exit(1);
        }
    }
    void WeakUpEventfd() // 唤醒eventfd
    {
        uint64_t val = 1;
        int ret = write(_eventfd, &val, sizeof(val));
        if (ret < 0)
        {
            if (errno == EINTR)
                return;
            LOG_ERROR("eventfd write error");
            exit(1);
        }
    }

public:
    EventLoop() : _id(std::this_thread::get_id()), _eventfd(CreateEventfd()),
                  _eventfd_channel(new Channel(_eventfd, this)), _time_wheel(this)
    {
        _eventfd_channel->SetReadAction(std::bind(&EventLoop::ReadEventfd, this));
        _eventfd_channel->EnableRead();
    }

    bool IsInLoop() // 判断当前线程是否是eventloop所对应的线程
    {
        return (_id == std::this_thread::get_id());
    }

    void AssertInLoop()
    {
        assert(_id == std::this_thread::get_id());
    }

    void RunInLoop(const _Func &_func) // 当要执行的任务处在当前线程中,则直接执行 否则将其压入任务池
    {
        if (IsInLoop())
            _func();
        else
            QueueInLoop(_func);
    }

    void QueueInLoop(const _Func &_func) // 将任务压入到任务池中
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(_func);
        }

        WeakUpEventfd(); // 唤醒eventfd
    }

    void UpdateEvents(Channel *_channel) // 更新描述符监控事件
    {
        _poller.UpdateEvents(_channel);
    }

    void RemoveEvents(Channel *_channel) // 移除描述符监控
    {
        _poller.RemoveEvents(_channel);
    }

    void Start()
    {
        // 1.事件监控
        while (1)
        {
            std::vector<Channel *> _actives;
            _poller.Poll(&_actives);
            // 2.事件处理
            for (auto &e : _actives)
            {
                e->EventAction();
            }
            // 3.执行任务
            RunAllTask();
        }
    }

    void TimerTaskAdd(uint64_t _id, uint32_t _timeout, const Task &_task)
    {
        _time_wheel.TimerTaskAddInLoop(_id, _timeout, _task);
    }

    void TimerTaskRefresh(uint64_t _id)
    {
        _time_wheel.TimerTaskRefreshInLoop(_id);
    }

    bool HasTimer(uint64_t _id)
    {
        return _time_wheel.HasTimer(_id);
    }

    void TimerCancel(uint64_t _id)
    {
        _time_wheel.TimerCancelInLoop(_id);
    }

private:
    std::thread::id _id;                       // 线程id
    std::vector<_Func> _tasks;                 // 任务池
                           // 描述符监控
    int _eventfd;                              // 用于唤醒线程
    std::unique_ptr<Channel> _eventfd_channel; // 用于管理_eventfd的监控事件
    Poller _poller; 
    std::mutex _mutex;                         // 用于保证任务池的线程安全性
    TimeWheel _time_wheel;                     // 时间轮
};

class LoopThread
{
public:
    LoopThread() : _loop(nullptr), _thread(std::thread(&LoopThread::ThreadAction, this))
    {
    }

    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _cond.wait(_lock, [this]() -> bool
                       { return _loop != nullptr; });
            loop = _loop;
        }

        return loop;
    }

private:
    void ThreadAction()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }

private:
    std::mutex _mutex;
    std::condition_variable _cond; // 互斥锁和条件变量保证EventLoop先初始化再获得
    EventLoop *_loop;
    std::thread _thread;
};

void Channel::Remove()
{
    _loop->RemoveEvents(this);
}

void Channel::Update()
{
	if (_loop == nullptr) LOG_INFO("_LOOP 是空的\n");
    _loop->UpdateEvents(this);
}

void TimeWheel::TimerTaskAddInLoop(uint64_t _id, uint32_t _timeout, const Task &_task)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerTaskAdd, this, _id, _timeout, _task));
}

void TimeWheel::TimerTaskRefreshInLoop(uint64_t _id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerTaskRefresh, this, _id));
}

void TimeWheel::TimerCancelInLoop(int _id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerCancel, this, _id));
}

typedef enum
{
    CONNECTED,
    CONNECTING,
    DISCONNECTED,
    DISCONNECTING
} Connect_Status;

class Connection;
using PtrConnection = std::shared_ptr<Connection>;
using ConnectionCallBack = std::function<void(const PtrConnection &)>;
using MessageCallBack = std::function<void(const PtrConnection &, Buffer *)>;
using CloseCallBack = std::function<void(const PtrConnection &)>;
using AnyEventCallBack = std::function<void(const PtrConnection &)>;

class Connection : public std::enable_shared_from_this<Connection>
{
private:
    void HandlerRead() // 给channel设置可读事件回调函数,并调用MessageCallBack
    {
        char buffer[65536];
        int _ret = _socket.NonBlockRecv(buffer, sizeof(buffer));
        if (_ret < 0)
        {
            ShutDownInLoop();
            return;
        }
        _inbuffer.WriteAndPush(buffer, _ret);
        if (_inbuffer.ReadbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_inbuffer);
            return;
        }
    }
    void HandlerWrite() // channel 可写事件回调函数
    {
        ssize_t ret = _socket.NonBlockSend(_outbuffer.ReadPosition(), _outbuffer.ReadbleSize());
        if (ret < 0)
        {
            if (_inbuffer.ReadbleSize() > 0)
            {
                _message_callback(shared_from_this(), &_inbuffer);
            }
            return Release();
        }
        _outbuffer.MoveReadOffest(ret);
        if (_outbuffer.ReadbleSize() == 0)
        {
            _channel.DisableWrite();
            if (_status == DISCONNECTING)
            {
                Release();
                return;
            }
        }
        return;
    }
    void HandlerClose() // channel 关闭事件回调函数
    {
        if (_inbuffer.ReadbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_inbuffer);
        }
        return Release();
    }
    void HandlerError() // channel 错误事件回调函数
    {
        return HandlerClose();
    }
    void HanlerEvent() // channel 触发任意事件回调函数
    {
        if (Enable_Inactive_Destroy)
            _loop->TimerTaskRefresh(_ConnId);
        if (_any_callback)
            _any_callback(shared_from_this());
    }

    void SendInLoop(char *data, size_t _len)
    {
        if (_status == DISCONNECTED)
            return;
        _outbuffer.WriteAndPush(data, _len);
        if (_channel.WriteAble() == false)
            _channel.EnableWrite();
    }
    void EstablishInLoop()
    {
        assert(_status == CONNECTING);
        _status = CONNECTED;
        _channel.EnableRead();
        if (_connection_callback)
            _connection_callback(shared_from_this());
    }
    void ShutDownInLoop()
    {
        _status = DISCONNECTING;
        if (_inbuffer.ReadbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_inbuffer);
        }

        if (_outbuffer.ReadbleSize() > 0)
        {
            if (_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }

        if (_outbuffer.ReadbleSize() == 0)
        {
            Release();
        }
    }
    void ReleaseInLoop()
    {
        _status = CONNECTED;
        _channel.Remove();
        _socket.Close();

        if (_loop->HasTimer(_ConnId))
            CancelInactiveDestroryInLoop();
        if (_close_callback)
            _close_callback(shared_from_this());
        if (_server_close_callback)
            _server_close_callback(shared_from_this());
    }
    void EnableInactiveDestroyInLoop(int _sec)
    {
        Enable_Inactive_Destroy = true;
        if (_loop->HasTimer(_ConnId))
            return _loop->TimerTaskRefresh(_ConnId);

        _loop->TimerTaskAdd(_ConnId, _sec, std::bind(&Connection::Release, this));
    }
    void CancelInactiveDestroryInLoop()
    {
        Enable_Inactive_Destroy = false;
        if (_loop->HasTimer(_ConnId))
            _loop->TimerCancel(_ConnId);
    }
    void SwitchProtrolInLoop(const std::any &context, const ConnectionCallBack &connection_callback,
                             const MessageCallBack &message_callback,
                             const CloseCallBack &close_callback,
                             const AnyEventCallBack &any_callback)
    {
        _context = context;
        _connection_callback = connection_callback;
        _message_callback = message_callback;
        _close_callback = close_callback;
        _any_callback = any_callback;
    }

public:
    Connection(EventLoop *loop, uint64_t id, int sockfd) 
        : _ConnId(id), Enable_Inactive_Destroy(false)
        ,_loop(loop), _sockfd(sockfd), 
        _socket(_sockfd), _status(CONNECTING)
        , _channel(_sockfd, _loop)
    {
		if (_loop == nullptr) LOG_INFO("connection --> loop is null");
        _channel.SetAnyAction(std::bind(&Connection::HanlerEvent, this));
        _channel.SetCloseAction(std::bind(&Connection::HandlerClose, this));
        _channel.SetErrorAction(std::bind(&Connection::HandlerError, this));
        _channel.SetReadAction(std::bind(&Connection::HandlerRead, this));
        _channel.SetWriteAction(std::bind(&Connection::HandlerWrite, this));
    }
    ~Connection()
    {
    }

    void Send(void *data, size_t _len) // 发送数据
    {
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, (char *)data, _len));
    }
    void ShutDown() // 关闭连接
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    void Release()
    {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop,this));
    }

    void EnableInactiveDestroy(int _sec) // 启动非活跃连接销毁 设置定时任务
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveDestroyInLoop, this, _sec));
    }
    void CancelInactiveDestroy()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveDestroryInLoop, this));
    } // 取消非活跃连接销毁
    void Establish()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishInLoop, this));
    }
    void SwitchProtrol(const std::any &context, const ConnectionCallBack &connection_callback,
                       const MessageCallBack &message_callback,
                       const CloseCallBack &close_callback,
                       const AnyEventCallBack &any_callback) // 切换协议
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::SwitchProtrolInLoop, this, context, connection_callback,
                                   message_callback, close_callback, any_callback));
    }
    int GetFd() // 获取所管理的文件描述符
    {
        return _sockfd;
    }
    int GetId() // 获取连接id
    {
        return _ConnId;
    }
    bool Connected() // 判断当前连接是否处于connected
    {
        return _status == CONNECTED;
    }
    void SetContext(const std::any &context) // 设置上下文
    {
        _context = context;
    }
    std::any *GetContext()
    {
        return &_context;
    } // 获取上下文
    void SetConnectionCallBack(const ConnectionCallBack &connect_callback)
    {
        _connection_callback = connect_callback;
    }
    void SetMessageCallBack(const MessageCallBack &message_callback)
    {
        _message_callback = message_callback;
    }
    void SetCloseCallBack(const CloseCallBack &close_callback)
    {
        _close_callback = close_callback;
    }
    void SetAnyEventCallBack(const AnyEventCallBack &any_callback)
    {
        _any_callback = any_callback;
    }

    void SetServerCloseCallback(const CloseCallBack &server_close_callback)
    {
        _server_close_callback = server_close_callback;
    }

private:
    uint64_t _ConnId;             // 每个连接所对应的id,便于连接的查找和管理
    int _sockfd;                  // 连接所关系的文件描述符
    Socket _socket;               // 套接字操作管理
    EventLoop *_loop;             // 连接所关联的eventloop             找到了,这俩成员定义反了,应该先定义loop,在定义channel,因为channel的初始化依赖loop
    Channel _channel;             // 对连接事件管理
    bool Enable_Inactive_Destroy; // 是否启动非活跃连接的销毁
    Connect_Status _status;       // 连接当前所对应的状态
    Buffer _inbuffer;             // 输入缓冲区
    Buffer _outbuffer;            // 输出缓冲区
    std::any _context;            // 请求的处理上下文
    ConnectionCallBack _connection_callback;
    MessageCallBack _message_callback;
    CloseCallBack _close_callback;
    AnyEventCallBack _any_callback;
    CloseCallBack _server_close_callback;
};

using Acceptcb = std::function<void(int)>;

class Acceptor
{
private:
    int CreateServer(int port)
    {
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.GetFd();
    }

    void HandlerRead()
    {
        int _newfd = _socket.Accept();
        if (_newfd < 0)
            return;

        if (_accept_cb)
            _accept_cb(_newfd);
    }

public:
    Acceptor(EventLoop *loop, int port) : _socket(CreateServer(port)), _loop(loop), _channel(_socket.GetFd(), _loop)
    {
        _channel.SetReadAction(std::bind(&Acceptor::HandlerRead, this));
    }

    void SetAccpetcb(const Acceptcb &_cb)
    {
        _accept_cb = _cb;
    }

    void Listen()
    {
        _channel.EnableRead();
    }

    ~Acceptor()
    {
    }

private:
    Socket _socket;      // 监听套接字
    EventLoop *_loop;    // 对监听套接字进行监控
    Channel _channel;    // 对监听套接字进行事件管理
    Acceptcb _accept_cb; // 对新连接的处理由上层决定,这里只提供回调
};

class LoopThreadPool
{

public:
    LoopThreadPool(EventLoop *baseLoop) : _threads_count(0), _nextLoop_ids(0), _baseLoop(baseLoop)
    {
    }

    void SetThreadsCount(int count)
    {
        _threads_count = count;
    }

    void Create()
    {
        if (_threads_count > 0)
        {
            _threads.resize(_threads_count);
            _Eventloops.resize(_threads_count);

            for (int i = 0; i < _threads_count; i++)
            {
                _threads[i] = new LoopThread();
                _Eventloops[i] = _threads[i]->GetLoop();
                if (_Eventloops[i] == nullptr) {
                    LOG_INFO("_Eventloops %d 是空.......", i);
                }
            }
        }

        return;
    }

    EventLoop *GetNextLoop()
    {
        if (_threads_count == 0)
        {
            return _baseLoop;
        }

        _nextLoop_ids = (_nextLoop_ids + 1) % _threads_count;

        return _Eventloops[_nextLoop_ids];
    }

private:
    int _threads_count; // 从属线程的数量
    int _nextLoop_ids;
    EventLoop *_baseLoop;                 // 主eventloop 当从属线程数量为0时,用baseloop处理事务
    std::vector<LoopThread *> _threads;   // 管理从属线程
    std::vector<EventLoop *> _Eventloops; // 管理所有的EventLoop
};

class TcpServer
{
private:
    int _port;                                          // server所对应的端口
    uint64_t _next_id;                                  // 连接id 用于对连接进行查找和管理
    int _timeout;                                       // 定义连接超时时间
    bool _enable_inactive_destroy;                      // 是否启动非活跃连接的超时销毁功能
    EventLoop _base_loop;                               // 主eventloop 负责对监听套接字进行事件监控 也就是获取新连接
    Acceptor _acceptor;                                 // 监听套接字事件管理对象
    std::unordered_map<uint64_t, PtrConnection> _conns; // 对所有连接进行管理
    LoopThreadPool _pool;                               // 从属线程池
    ConnectionCallBack _connection_callback;
    MessageCallBack _message_callback;
    CloseCallBack _close_callback;
    AnyEventCallBack _any_callback;

private:
    void NewConnection(int fd) // 对新建连接的操作
    {

		LOG_INFO("INTO NewConnection--------");
        _next_id++;
		auto loop = _pool.GetNextLoop();
		if (loop == nullptr) LOG_INFO("_pool.GetNextLoop() IS NULL");
        PtrConnection conn(new Connection(_pool.GetNextLoop(), _next_id, fd));
        conn->SetMessageCallBack(_message_callback);
        conn->SetCloseCallBack(_close_callback);
        conn->SetConnectionCallBack(_connection_callback);
        conn->SetAnyEventCallBack(_any_callback);
        conn->SetServerCloseCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if (_enable_inactive_destroy)
            conn->EnableInactiveDestroy(_timeout); // 启动非活跃超时销毁
        conn->Establish();                         // 就绪初始化
        _conns.insert(std::make_pair(_next_id, conn));
    }

    void RemoveConnectionInLoop(const PtrConnection &Conn)
    {
        int _id = Conn->GetId();
        auto it = _conns.find(_id);
        if (it != _conns.end())
        {
            _conns.erase(it);
        }
    }

    void RemoveConnection(const PtrConnection &Conn) // 从_conns中移除连接
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, Conn));
    }

    void RunAfterInLoop(const Task &_task, int delay) // 用于添加一个定时任务
    {
        _next_id++;
        _base_loop.TimerTaskAdd(_next_id, delay, _task);
    }

public:
    TcpServer(int port)
        : _port(port), _next_id(0), _enable_inactive_destroy(false), _acceptor(&_base_loop, _port),
          _pool(&_base_loop)
    {
        _acceptor.SetAccpetcb(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen();
    }

    void SetConnectionCallBack(const ConnectionCallBack &connect_callback)
    {
        _connection_callback = connect_callback;
    }
    void SetMessageCallBack(const MessageCallBack &message_callback)
    {
        _message_callback = message_callback;
    }
    void SetCloseCallBack(const CloseCallBack &close_callback)
    {
        _close_callback = close_callback;
    }
    void SetAnyEventCallBack(const AnyEventCallBack &any_callback)
    {
        _any_callback = any_callback;
    }

    void SetThreadsCounts(int counts)
    {
        _pool.SetThreadsCount(counts);
    }

    void RunAfter(const Task &_task, int delay) // 用于添加一个定时任务
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, _task, delay));
    }

    void Start()
    {
        _pool.Create();
        _base_loop.Start();
    }

    void EnableInactiveRelease(int timeout)
    {
        _timeout = timeout;
        _enable_inactive_destroy = true;
    }
};
